package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.common.article.mess.ArticleVisitStreamMess;
import com.heima.model.common.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    //@Bean
    public KStream<String,String > kStream(StreamsBuilder streamsBuilder){
        //1.接收消息
        KStream<String, String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        kStream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //1.接收消息
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                /**
                 *  2.重新定义key和value的值
                 *  key=文章id,value=类型：增量
                 */
                return new KeyValue<String,String>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
            }
        })
                //以key进行分组聚合
                .groupBy((key,value)-> key)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //聚合计算
                .aggregate(new Initializer<String>() {
                    /**
                     * 初始化聚合，返回值是消息的value
                     * 初始化，COLLECTION收藏分为0，COMMENT评论分为0，LIKES点赞分为0，VIEWS阅读分为0
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }
                    //正式聚合计算
                    , new Aggregator<String, String, String>() {
                            /**
                             * 进行分值统计
                             * @param key key值  文章id
                             * @param value value值   类型:增量值 like:1
                             * @param init  初始化返回值， COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                             * @return
                             */
                    @Override
                    public String apply(String key, String value, String init) {
                        //如果value值为空
                        if(StringUtils.isEmpty(value)){
                            return init;//直接返回初始化值
                        }
                        //定义变量
                         int collection = 0,comment=0,like=0,view=0;
                        //字符串转化成数组
                        String[] aggArray = init.split(",");//[COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0]
                        for (String agg : aggArray) {//COLLECTION:0
                            String[] split = agg.split(":");
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                case COLLECTION:
                                    collection=Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    comment=Integer.parseInt(split[1]);
                                    break;
                                case LIKES:
                                    like=Integer.parseInt(split[1]);
                                    break;
                                case VIEWS:
                                    view=Integer.parseInt(split[1]);
                                    break;
                            }
                        }
                        /**
                         * 累加操作，从value中取出分值，然后累加到变量中
                         */
                        String[] split = value.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                            case COLLECTION:
                                collection +=Integer.parseInt(split[1]);
                                break;
                            case COMMENT:
                                comment +=Integer.parseInt(split[1]);
                                break;
                            case LIKES:
                                like +=Integer.parseInt(split[1]);
                                break;
                            case VIEWS:
                                view +=Integer.parseInt(split[1]);
                                break;
                        }

                        //返回类型：分值
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", collection, comment, like, view);
                        System.out.println("文章的id:"+key);
                        System.out.println("当前时间窗口内的消息处理结果："+formatStr);
                        return formatStr;
                    }
                },Materialized.as("hot-article_stream"))
                .toStream()
                .map((key,value)-> {
                    System.out.println("key值:"+key.key().toString());
                    System.out.println("value值："+value.toString());

                    return new KeyValue<String ,String >(key.key().toString(),formatMess(key.key().toString(),value.toString()));
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return kStream;
    }

    /**
     * 格式化value值，然后再发送消息
     * 把COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0格式化成ArticleVisitStreamMess类型
     * @param key 文章id
     * @param value  COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     * @return
     */
    private String formatMess(String key, String value) {
        ArticleVisitStreamMess mess=new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(key));

        String[] aggArray = value.split(",");
        for (String agg : aggArray) {
            String[] split = agg.split(":");
        switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
            case COLLECTION:
                mess.setCollect(Integer.parseInt(split[1]));
                break;
            case COMMENT:
                mess.setComment(Integer.parseInt(split[1]));
                break;
            case LIKES:
                mess.setLike(Integer.parseInt(split[1]));
                break;
            case VIEWS:
               mess.setView(Integer.parseInt(split[1]));
                break;
        }
        }
        log.info("聚合消息处理之后的结果为:{}",JSON.toJSONString(mess));
        return JSON.toJSONString(mess);
    }
}